package com.atguigu.hbase;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * Created by Smexy on 2022/10/5
 *
 *  所有的客户端，只要要连接hbase集群，都需要知道hbase集群所使用的zk集群的地址！
 *
 *  ①创建一个客户端对象
 *          Connection代表客户端和服务端的连接。
 *          可以通过ConnectionFactory创建。
 *                  如何创建对象。
 *                      传统：
 *                             A a = new A()
 *                      设计模式:
 *                          建造者模式: A a = new ABuilder().build()
 *                          工厂模式： A a =  new AFactory().getInstance()
 *          Connection创建起来很费劲，建议一个App只创建一个，可以在不同的线程之间分享。
 *
 *          从Connection中获取
 *              Table:  执行 DML，对数据的增删改查
 *              Admin： 执行 DDL， 建表，删表等
 *              Table和Admin的创建是轻量级(不费劲)，不建议池化，或缓存，而是每个线程有自己的table，用完就关掉。
 *  ②连接服务端
 *  ③使用客户端发送命令
 *          对哪张表操作，需要创建一个对应的Table。
 *
 *          增改： Table.put(Put p)
 *          删:  Table.delete(Delete d)
 *          查:  Table.get(Get g)
 *               Table.scan(Scan s)
 *
 *
 *       两个工具类:
 *          Bytes:     把类型转为byte[]:      Bytes.toBytes(xxx)
 *                      把byte[] 转为xxx类型: Bytes.toXxx(byte [] b)
 *
 *          CellUtil:   获取一个Cell中的属性。
 *                          获取一个cell的rowkey:   CellUtil.cloneRowkey()
 *
 *  ④如果是读操作，接收返回的结果
 *  ⑤关闭客户端
 */
public class HBaseUtil
{

    //创建一个单例Connection
    private static Connection conn;

    //封装Put(一个Put代表对一个Cell的操作)
    public static Put getPut(String rowkey,String cf,String cq,String value){

        //向某一行去插入
        Put put = new Put(Bytes.toBytes(rowkey));

        //每调用一次addColumn是插入一个cell
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes(cq),Bytes.toBytes(value));

        return put;

    }

    //解析一行的所有列，在控制台打印下
    public static void parseResult(Result result){

        //获取一行所有的cell
        Cell[] cells = result.rawCells();

        for (Cell cell : cells) {

            System.out.println("row:"
                + Bytes.toString(CellUtil.cloneRow(cell)) + "   "
                + Bytes.toString(CellUtil.cloneFamily(cell)) + ":"
                + Bytes.toString(CellUtil.cloneQualifier(cell)) + "  "
                + Bytes.toString(CellUtil.cloneValue(cell)));

        }

    }


    //返回一个表的Table对象
    public static Table getTable(String tableName) throws IOException {

        //判断表名是否是白字符(空字符串，回车，tab)或null
        if (StringUtils.isBlank(tableName)){
            throw new RuntimeException("表名非法!");
        }

        return conn.getTable(TableName.valueOf(tableName));

    }

    static {

        //自动读取类路径下hbase-site.xml中的配置
        try {
            conn = ConnectionFactory.createConnection();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void closeConn(){
        if (conn != null){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
